package com.geostar.constructionland.statistics.service;

import com.geostar.constructionland.common.entity.statistics.LogStatistics;
import com.geostar.constructionland.common.entity.statistics.StatisticsProfile;
import com.geostar.constructionland.statistics.enumeration.StatisticStatus;
import com.geostar.constructionland.statistics.enumeration.TaskStatus;
import com.geostar.constructionland.statistics.util.RedisLock;
import com.geostar.constructionland.statistics.util.TaskListUtil;
import com.geostar.constructionland.statistics.util.TaskObject;
import com.geostar.constructionland.statistics.util.VersionCompareUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * @author yelll
 * @since 2020-7-14 10:44
 **/
@Service
@Slf4j
public class StatisticsService {

    /**
     * 超时时间 5s
     */
    private static final String TIMEOUT = "5000";
    /**
     * 统计业务全局分布式锁 键值
     */
    private static final String LOCK_SIGNATURE = "STATISTIC:DISTRIBUTED:SIGNATURE";
    @Autowired
    private StatisticsProfileService statisticsProfileService;
    @Autowired
    private AsyncConsumerService asyncConsumerService;
    @Autowired
    private LogStaticService logStaticService;
    @Autowired
    private RedisLock redisLock;

    /**
     * @param uniqueTaskId 分析类型
     * @param ywid         业务id
     * @return
     */
    public String getStatistic(String uniqueTaskId, String ywid) {
        // 1. 根据fxlx和ywid确定一组统计配置实体类，根据版本后筛选出最新的一个
        List<StatisticsProfile> statisticsProfileList = statisticsProfileService.getStatisticsProfile(uniqueTaskId);
        if (statisticsProfileList.size() == 0) {
            return "Wrong Arguments,Can not find Statistics Profile";
        }

        StatisticsProfile statisticsProfile =
                statisticsProfileList.stream()
                        // 按照版本号排序，最新的放在List最前
                        .sorted((profile1, profile2) -> VersionCompareUtil.compareVersion(profile2.getStatisticsVersion(), profile1.getStatisticsVersion()))
                        .findFirst()
                        .get();

        // 如果不包含任何参数则拒绝执行，避免恶意删库
        if(!statisticsProfile.getExecuteSql().contains(":ywid")) {
            return "You must specify the parameter ywid";
        }

        String replaceSql = statisticsProfile.getExecuteSql().replaceAll(":ywid", "'" + ywid + "'");
        statisticsProfile.setExecuteSql(replaceSql);

        // 2. 获取全局的唯一任务队列 模拟去Redis中读取任务队列过程
        LinkedList<TaskObject> taskList = TaskListUtil.getTaskList();

        // 3. 模拟生产者获取分布式锁 查找是否存在当前任务，没有则添加并将任务插入数据库，有则抛弃新任务（确保队列里的统计任务的唯一性）
        // 重写了contains方法，避免同一种任务反复添加
        redisLock.lock(LOCK_SIGNATURE, TIMEOUT);
        int index = this.containUnconsumedTask(taskList, statisticsProfile);
        if (index >= 0) {
            TaskObject removedTask = taskList.remove(index);
            taskList.addFirst(removedTask);
        } else {
            // 生成执行记录 插入到数据库
            LogStatistics statistics = new LogStatistics();
            statistics.setFxlx(statisticsProfile.getFxlx())
                    .setYwid(ywid)
                    .setProfileId(statisticsProfile.getId())
                    .setStatus(StatisticStatus.NOT_YET.getValue())
                    .setModifiedTime(new Date());
            logStaticService.addEntity(statistics);

            TaskObject taskObject = TaskObject.builder().task(statisticsProfile).taskStatus(TaskStatus.INIT).other(statistics).build();
            taskList.addLast(taskObject);
        }
        redisLock.release(LOCK_SIGNATURE, TIMEOUT);


        // 4. 模拟消费者获取分布式锁，获取第一个任务，将其状态改为已消费，释放分布式锁
        TaskObject task = null;
        redisLock.lock(LOCK_SIGNATURE, TIMEOUT);
        task = this.getFirstInitTask(taskList);
        if (!ObjectUtils.isEmpty(task)) {
            task.setTaskStatus(TaskStatus.CONSUMED);
        }
        redisLock.release(LOCK_SIGNATURE, TIMEOUT);

        // 5. 任务提交给新的消费者线程处理，立即返回
        CompletableFuture<String> futureResult = asyncConsumerService.generateStatisticTable(task);
        asyncConsumerService.handleResult(futureResult);

        // 6. 清理任务队列中已完成的任务
        redisLock.lock(LOCK_SIGNATURE, TIMEOUT);
        clearAllCompleted(taskList);
        redisLock.release(LOCK_SIGNATURE, TIMEOUT);

        return "SUCCESS";
    }

    // 判断队列中是否存在未被消费的相同任务（防止重复提交任务到任务队列）
    private int containUnconsumedTask(LinkedList<TaskObject> list, StatisticsProfile statisticsProfile) {
        for (int i = 0; i < list.size(); i++) {
            TaskObject taskObject = list.get(i);
            // 队列中存在相同任务且任务未被完成，则返回该任务的索引
            if (taskObject.getTask().equals(statisticsProfile) && !TaskStatus.COMPLETED.equals(taskObject.getTaskStatus())) {
                return i;
            }
        }
        return -1;
    }

    private TaskObject getFirstInitTask(LinkedList<TaskObject> list) {
        for (int i = 0; i < list.size(); i++) {
            TaskObject taskObject = list.get(i);
            if (TaskStatus.INIT.equals(taskObject.getTaskStatus())) {
                return taskObject;
            }
        }
        return null;
    }

    private void clearAllCompleted(LinkedList<TaskObject> list) {
        Iterator<TaskObject> iterator = list.iterator();
        while (iterator.hasNext()) {
            TaskObject taskObject = iterator.next();
            if (TaskStatus.COMPLETED.equals(taskObject.getTaskStatus())) {
                iterator.remove();
            }
        }
    }
}
